package job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class CarAppLogEtlJob {
    public static void main(String[] args) throws Exception {
        // 1.env执行环境
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        // 2.消费kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle_realtime");
        kafkaStream.print("Kafka");
        // 3.数据转换
        DataStream<String> processKafkaStream = processLog(kafkaStream);
        // 4.消费到kafka
        KafkaUtil.producerKafka(processKafkaStream,"dwd-car-logs");
        // 5.触发执行
        env.execute("CarAppLogEtlJob");
    }

    private static DataStream<String> processLog(DataStream<String> stream) {
        //数据清洗
        DataStream<String> dataCleanStream = DataClean(stream);
        //数据过滤
        DataStream<String> dataFilterStream = dataFilter(dataCleanStream);
        return dataFilterStream;
    }

    private static DataStream<String> dataFilter(DataStream<String> stream) {
        DataStream<String> dataFilterSteam = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String vin = jsonObject.getString("vin");
                Long timestamp = jsonObject.getLong("timestamp");
                Double longitude = jsonObject.getDoubleValue("longitude");
                Double latitude = jsonObject.getDoubleValue("latitude");
                if (vin == null || vin.isEmpty() || timestamp == null || longitude == null || latitude == null) {
                    return false;
                }
                return true;
            }
        });
        return dataFilterSteam;
    }

    /**
     * 数据清洗:对非JSON格式的数据进行侧边流输出,JSON格式数据主流输出
     * @param stream
     * @return
     */
    private static DataStream<String> DataClean(DataStream<String> stream) {
        final OutputTag<String> dirtyTag = new OutputTag<String>("dirty-Tag"){};
        SingleOutputStreamOperator<String> cleanData = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    // 解析JSON数据
                    JSON.parseObject(value);
                    // 解析成功主流输出
                    out.collect(value);
                } catch (Exception e) {
                    //解析失败存入测流
                    ctx.output(dirtyTag, value);
                }
            }
        });
        SideOutputDataStream<String> dirtyLog = cleanData.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirtyLog,"car-dirty-logs");
        return cleanData;
    }
}
